package org.apache.ocean.main;

import java.io.IOException;
import java.net.URL;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
import java.util.Set;

import org.apache.lucene.analysis.Analyzer;
import org.apache.lucene.index.CorruptIndexException;
import org.apache.lucene.index.IndexReader;
import org.apache.lucene.index.StaleReaderException;
import org.apache.lucene.index.Term;
import org.apache.lucene.index.TermDocs;
import org.apache.lucene.search.DocIdSetIterator;
import org.apache.lucene.search.HitCollector;
import org.apache.lucene.search.IndexSearcher;
import org.apache.lucene.search.Query;
import org.apache.lucene.util.OpenBitSet;
import org.apache.ocean.main.Deletes.DeleteByQuery;
import org.apache.ocean.main.TransactionSystem.Category;
import org.apache.ocean.main.fields.SchemaField;
import org.apache.ocean.main.search.FieldCursorImpl;
import org.apache.ocean.main.search.cache.Cache;
import org.apache.ocean.main.search.cache.CacheKey;
import org.apache.ocean.main.search.cache.DocSetCacheKey;
import org.apache.ocean.main.search.cache.FacetCacheKey;
import org.apache.ocean.main.search.cache.FieldCacheKey;
import org.apache.ocean.main.search.cache.FieldCacheKey.FieldIndex;
import org.apache.ocean.main.util.BitDocSet;
import org.apache.ocean.main.util.MutableInt;
import org.apache.ocean.main.util.Range;
import org.apache.ocean.main.util.Timeout;

// TODO: implement range filter using field cache value comparisons
public abstract class Index {
	private final Long id;
	protected final Analyzer defaultAnalyzer;
	protected boolean isClosed = false;
	private final TransactionSystem.Category category;
	private boolean isReadOnly = false;
	private boolean isDeleteOnly = false;

	public Index(Long id, Analyzer defaultAnalyzer, TransactionSystem.Category category) {
		this.id = id;
		this.defaultAnalyzer = defaultAnalyzer;
		this.category = category;
	}
  
	protected Cache initCache() {
		Category.Config config = category.getConfig();
		int docSetMax = config.getDocSetCacheMax(100);
		return new Cache(docSetMax);
	}
	
	public static IndexReader[] getIndexReaders(List<? extends IndexSnapshot> indexSnapshots) {
		IndexReader[] indexReaders = new IndexReader[indexSnapshots.size()];
		for (int x = 0; x < indexSnapshots.size(); x++) {
			indexReaders[x] = indexSnapshots.get(x).getIndexReader();
		}
		return indexReaders;
	}

	public static Long getHighestSnapshotId(List<? extends IndexSnapshot> indexSnapshots) throws Exception {
		List<Long> snapshotIdList = new ArrayList<Long>(indexSnapshots.size());
		for (IndexSnapshot indexSnapshot : indexSnapshots) {
			snapshotIdList.add(indexSnapshot.getHighestSnapshotId());
		}
		return Collections.max(snapshotIdList);
	}

	public static Long getHighestDocumentId(List<? extends IndexSnapshot> indexSnapshots) throws Exception {
		List<Long> documentIdList = new ArrayList<Long>(indexSnapshots.size());
		for (IndexSnapshot indexSnapshot : indexSnapshots) {
			documentIdList.add(indexSnapshot.getHighestDocumentId());
		}
		return Collections.max(documentIdList);
	}

	public static boolean hasAllFieldIndices(IndexSchema.Category schemaCategory, Cache cache) {
		for (SchemaField schemaField : schemaCategory.getFieldIndexables()) {
			FieldCacheKey fieldCacheKey = new FieldCacheKey(schemaField.getName(), schemaField.getFieldType().getType());
			if (!cache.containsKey(fieldCacheKey)) {
				return false;
			}
		}
		return true;
	}

	public TransactionSystem.Category getCategory() {
		return category;
	}

	public void setDeleteOnly(boolean isDeleteOnly) {
		this.isDeleteOnly = isDeleteOnly;
	}

	public boolean isDeleteOnly() {
		return isDeleteOnly;
	}

	public boolean isReadOnly() {
		return isReadOnly;
	}

	public void setReadOnly(boolean isReadOnly) {
		this.isReadOnly = isReadOnly;
	}

	protected DeletesResult applyDeletes(boolean deleteFromReader, Deletes deletes, Collection<Integer> deletedDocs, IndexReader indexReader, FieldIndex idFieldIndex) throws CorruptIndexException, IOException, Exception {
		DeletesResult deletesResult = new DeletesResult(getId());
		if (deletes.hasDeletes()) {
			if (deletes.hasDocIds()) {
				int docsDeleted = 0;
				long[] docIdsArray = deletes.getDocIds();
				NavigableMap<Object, ValueIndex> sortedMap = idFieldIndex.getSortedMap();
				for (long docId : docIdsArray) {
					ValueIndex valueIndex = sortedMap.get(docId);
					if (valueIndex != null) {
						for (int doc : valueIndex.docs) {
							indexReader.deleteDocument(doc);
							docsDeleted++;
						}
					}
				}
				deletesResult.add(new DeletesResult.Result(docIdsArray, docsDeleted));
			} else {
				List<Long> docIds = deletesResult.getDocIds();
				if (deletes.hasTerms()) {
					List<Term> terms = deletes.getTerms();
					for (Term term : terms) {
						int docsDeleted = deleteByTerm(deleteFromReader, term, deletedDocs, docIds, idFieldIndex, indexReader);
						deletesResult.add(new DeletesResult.Result(term, docsDeleted));
					}
				}
				if (deletes.hasDeleteByQueries()) {
					List<DeleteByQuery> deleteByQueries = deletes.getDeleteByQueries();
					for (DeleteByQuery deleteByQuery : deleteByQueries) {
						int docsDeleted = deleteByQuery(deleteFromReader, deleteByQuery, deletedDocs, idFieldIndex, docIds, indexReader);
						deletesResult.add(new DeletesResult.Result(deleteByQuery, docsDeleted));
					}
				}
			}
		}
		return deletesResult;
	}

	protected int deleteByTerm(boolean deleteFromReader, Term term, Collection<Integer> deletedDocs, List<Long> docIds, FieldIndex idFieldIndex, IndexReader indexReader) throws IOException {
		TermDocs docs = indexReader.termDocs(term);
		if (docs == null)
			return 0;
		int n = 0;
		try {
			while (docs.next()) {
				int doc = docs.doc();
				if (deletedDocs != null)
					deletedDocs.add(doc);
				Long docId = (Long)idFieldIndex.getComparable(doc);
				docIds.add(docId);
				if (deleteFromReader)
					indexReader.deleteDocument(doc);
				n++;
			}
		} finally {
			docs.close();
		}
		return n;
	}

	protected int deleteByQuery(final boolean deleteFromReader, DeleteByQuery deleteByQuery, final Collection<Integer> deletedDocs, final FieldIndex idFieldIndex, final List<Long> docIds, final IndexReader indexReader) throws IOException {
		Query query = deleteByQuery.getQuery();
		IndexSearcher indexSearcher = new IndexSearcher(indexReader);
		final MutableInt numDeleted = new MutableInt(0);
		indexSearcher.search(query, new HitCollector() {
			public void collect(int doc, float score) {
				try {
					if (deleteFromReader)
						indexReader.deleteDocument(doc);
					if (deletedDocs != null)
						deletedDocs.add(doc);
					Long docId = (Long)idFieldIndex.getComparable(doc);
					docIds.add(docId);
					numDeleted.value++;
				} catch (StaleReaderException staleReaderException) {
					throw new RuntimeException(staleReaderException);
				} catch (IOException ioException) {
					throw new RuntimeException(ioException);
				}
			}
		});
		return numDeleted.value;
	}

	public static class DeletesResult {
		private long indexId;
		private List<Result> results = new ArrayList<Result>();
		private int numDeleted = 0;
		private List<Long> docIds = new LinkedList<Long>();

		public DeletesResult(long indexId) {
			this.indexId = indexId;
		}

		public List<Long> getDocIds() {
			return docIds;
		}

		public Long getIndexId() {
			return indexId;
		}

		public void add(Result result) {
			numDeleted += result.getNumDeleted();
			results.add(result);
		}

		public int getNumDeleted() {
			return numDeleted;
		}

		public static class Result {
			private Object delete;
			private int numDeleted;

			public Result(Object delete, int numDeleted) {
				this.delete = delete;
				this.numDeleted = numDeleted;
			}

			public Object getDelete() {
				return delete;
			}

			public int getNumDeleted() {
				return numDeleted;
			}
		}
	}

	public static class IndexException extends Exception {
		public IndexException(String message) {
			super(message);
		}

		public IndexException(String message, Exception exception) {
			super(message, exception);
		}
	}

	public static class IndexNeverCompletedCopyException extends IndexException {
		public IndexNeverCompletedCopyException(String message) {
			super(message);
		}
	}

	public IndexSchema.Category getIndexSchemaCategory() throws Exception {
		return category.getCategoryIndexSchema();
	}

	public abstract IndexSnapshot getIndexSnapshot(Long snapshotID);

	public abstract IndexSnapshot getLatestIndexSnapshot();

	public abstract class IndexSnapshot {
		protected final Long snapshotId;
		protected final IndexSchema.Category schemaCategory;

		public IndexSnapshot(Long snapshotId, IndexSchema.Category schemaCategory) {
			this.snapshotId = snapshotId;
			this.schemaCategory = schemaCategory;
		}

		public Long getSnapshotId() {
			return snapshotId;
		}

		public Index getIndex() {
			return Index.this;
		}

		public abstract FieldIndex createFieldIndex(String field, Timeout timeout) throws Exception;

		/**
		 * public FieldIndex createFieldIndex(String f, Timeout timeout) throws
		 * Exception { String field = f.intern(); FieldType fieldType =
		 * schemaCategory.getField(field).getFieldType(); IndexReader indexReader =
		 * getIndexReader(); final int[] retArray = new int[indexReader.maxDoc()];
		 * ValueIndex[] valueIndices = new ValueIndex[indexReader.maxDoc() + 1];
		 * TermDocs termDocs = indexReader.termDocs(); TermEnum termEnum =
		 * indexReader.terms(new Term(field, "")); int t = 1; // current term number //
		 * an entry for documents that have no terms in this field // should a
		 * document with no terms be at top or bottom? // this puts them at the top -
		 * if it is changed, FieldDocSortedHitQueue // needs to change as well. try {
		 * do { Term term = termEnum.term(); if (term == null || term.field() !=
		 * field) break; // store term text // we expect that there is at most one
		 * term per document if (t >= valueIndices.length) throw new
		 * RuntimeException("there are more terms than " + "documents in field \"" +
		 * field + "\", but it's impossible to sort on " + "tokenized fields");
		 * termDocs.seek(termEnum); List<Integer> list = new LinkedList<Integer>();
		 * while (termDocs.next()) { int doc = termDocs.doc(); retArray[doc] = t;
		 * list.add(doc); } int[] docs = Util.toIntArray(list); Object object =
		 * fieldType.stringToObject(term.text()); valueIndices[t] = new
		 * ValueIndex((Comparable)object, docs); t++; } while (termEnum.next()); }
		 * finally { termDocs.close(); termEnum.close(); } if (t == 0) { // if there
		 * are no terms, make the term array // have a single null entry
		 * valueIndices = new ValueIndex[1]; } else if (t < valueIndices.length) { //
		 * if there are less terms than documents, // trim off the dead array space
		 * ValueIndex[] array = new ValueIndex[t]; System.arraycopy(array, 0, array,
		 * 0, t); valueIndices = array; } List<Integer> nullDocList = new
		 * LinkedList<Integer>(); for (int x = 0; x < retArray.length; x++) { if
		 * (retArray[x] == 0) { nullDocList.add(x); } } valueIndices[0] = new
		 * ValueIndex(null, Util.toIntArray(nullDocList)); NavigableMap sortedMap =
		 * new TreeMap(); for (int x = 1; x < valueIndices.length; x++) {
		 * sortedMap.put(valueIndices[x].value, valueIndices[x]); } return new
		 * FieldIndex(retArray, valueIndices, sortedMap); }
		 */
    
		// TODO: implement somehow
		public ValueIndex getValueIndex(String field, Object value) throws Exception {
			FieldIndex fieldIndex = getFieldIndex(field);
			NavigableMap<Object, ValueIndex> sortedMap = fieldIndex.getSortedMap();
			return sortedMap.get(value);
		}

		public FieldCursor getFieldCursor(String field, Range range) throws Exception {
			return new FieldCursorImpl(field, range, this);
		}
		
		public abstract Integer getFacetValue(FacetCacheKey facetCacheKey) throws Exception;

		public BitDocSet createRangeDocSet(String field, Range range, Timeout timeout) throws Exception {
			FieldCursor fieldCursor = getFieldCursor(field, range);
			IndexReader indexReader = getIndexReader();
			int maxDoc = indexReader.maxDoc();
			OpenBitSet openBitSet = new OpenBitSet(maxDoc);
			do {
			  DocIdSetIterator docs = fieldCursor.docs();
			  while (docs.next()) {
			    openBitSet.fastSet(docs.doc());
			  }
			} while (fieldCursor.next());
			return new BitDocSet(openBitSet);
		}

		public IndexSchema.Category getIndexSchemaCategory() {
			return schemaCategory;
		}

		public abstract Long getHighestSnapshotId() throws Exception;

		public abstract Long getHighestDocumentId() throws Exception;

		public abstract Set<CacheKey> getCacheKeys();

		public abstract IndexReader getIndexReader();

		public abstract FieldIndex getFieldIndex(String field) throws Exception;

		public abstract BitDocSet getDocSet(DocSetCacheKey cacheKey, Timeout timeout) throws Exception;

		public abstract boolean containsKey(CacheKey cacheKey);
	}

	/**
	 * public abstract static class ResultFilter { public abstract boolean
	 * contains(int doc); }
	 * 
	 * public static class FieldCacheRangeSnapshotFilter { private final String
	 * field; private final Range range; private final IndexSnapshot
	 * indexSnapshot; private final Comparable[] array;
	 * 
	 * public FieldCacheRangeSnapshotFilter(String field, Range range,
	 * IndexSnapshot indexSnapshot) { this.indexSnapshot = indexSnapshot;
	 * this.field = field; this.range = range; array =
	 * indexSnapshot.getFieldArray(field); }
	 * 
	 * public ResultFilter getResultFilter() { if (range.isIncludeLower() &&
	 * range.isIncludeUpper()) { return new ResultFilter() { public boolean
	 * contains(int doc) { int lowerCompare =
	 * array[doc].compareTo(range.getLowerVal()); int upperCompare =
	 * array[doc].compareTo(range.getUpperVal()); if ((lowerCompare == 0 ||
	 * lowerCompare == 1) && (upperCompare == 0 || upperCompare == -1)) { return
	 * true; } return false; } }; } else if (range.isIncludeLower() &&
	 * !range.isIncludeUpper()) { return new ResultFilter() { public boolean
	 * contains(int doc) { int lowerCompare =
	 * array[doc].compareTo(range.getLowerVal()); int upperCompare =
	 * array[doc].compareTo(range.getUpperVal()); if ((lowerCompare == 0 ||
	 * lowerCompare == 1) && (upperCompare == -1)) { return true; } return false; } }; }
	 * else if (!range.isIncludeLower() && range.isIncludeUpper()) { return new
	 * ResultFilter() { public boolean contains(int doc) { int lowerCompare =
	 * array[doc].compareTo(range.getLowerVal()); int upperCompare =
	 * array[doc].compareTo(range.getUpperVal()); if ((lowerCompare == 1) &&
	 * (upperCompare == 0 || upperCompare == -1)) { return true; } return false; } }; }
	 * else { return new ResultFilter() { public boolean contains(int doc) { int
	 * lowerCompare = array[doc].compareTo(range.getLowerVal()); int upperCompare =
	 * array[doc].compareTo(range.getUpperVal()); if (lowerCompare == 1 &&
	 * upperCompare == -1) { return true; } return false; } }; } } }
	 */
	public static class MergedDocMap {
		private Map<IndexSnapshot, int[]> oldMap; // maps old doc to new doc
		private RI[] merged; // maps new doc to old doc and reader

		public MergedDocMap(List<? extends IndexSnapshot> indexSnapshots) {
			int newMaxDoc = 0;
			for (IndexSnapshot indexSnapshot : indexSnapshots) {
				newMaxDoc += indexSnapshot.getIndexReader().numDocs();
			}
			oldMap = new HashMap<IndexSnapshot, int[]>(indexSnapshots.size());
			RI[] merged = new RI[newMaxDoc];
			int pos = 0;
			for (IndexSnapshot indexSnapshot : indexSnapshots) {
				IndexReader indexReader = indexSnapshot.getIndexReader();
				int maxDoc = indexReader.maxDoc();
				int[] old = new int[maxDoc];
				for (int x = 0; x < maxDoc; x++) {
					if (indexReader.hasDeletions() && indexReader.isDeleted(x)) {
						merged[pos] = null;
						old[x] = -1;
					} else {
						merged[pos] = new RI(x, indexSnapshot);
						old[x] = pos;
						pos++;
					}
				}
				oldMap.put(indexSnapshot, old);
			}
		}

		public static class RI {
			public int doc;
			public IndexSnapshot oldIndexSnapshot;

			public RI(int doc, IndexSnapshot oldIndexSnapshot) {
				this.doc = doc;
				this.oldIndexSnapshot = oldIndexSnapshot;
			}
		}

		public Map<IndexSnapshot, int[]> getOldMap() {
			return oldMap;
		}

		public RI[] getMerged() {
			return merged;
		}
	}
	
	public abstract static class FieldCursor {
	  public abstract boolean next() throws Exception;
	  
	  public abstract DocIdSetIterator docs() throws Exception;
	  
	  public abstract Object value() throws Exception;
	  
	  public abstract void close() throws Exception;
	}
	
	/**
	 * for locking a disk index snapshot for index file replication
	 */
	public static class NodeRef {
		private String id;
		private URL url;

		public NodeRef(String id, URL url) {
			this.id = id;
			this.url = url;
		}

		public String getId() {
			return id;
		}

		public URL getUrl() {
			return url;
		}
	}

	public boolean isClosed() {
		return isClosed;
	}

	public Long getId() {
		return id;
	}

	public abstract DeletesResult commitDeletes(Deletes deletes, Transaction transaction) throws IndexException, InterruptedException, IOException;
}
